package cn.edu.flink.tutorial.window;


import cn.edu.flink.tutorial.wordcount.flatmap.WordCountFlatMap;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class TimeWindowWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
//        env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
//        DataStreamSource<String> text = env.readTextFile("src/main/resources/word.txt");
        DataStreamSource<String> text = env.socketTextStream("localhost", 8088);

        SingleOutputStreamOperator<String> sum =
                text.flatMap(new WordCountFlatMap())
                .keyBy(0)
                .timeWindow(Time.seconds(5))
                .process(new ProcessWindowFunction<Tuple2<String, Integer>, String, Tuple, TimeWindow>() {
                    @Override
                    public void process(Tuple tuple, ProcessWindowFunction<Tuple2<String, Integer>, String, Tuple, TimeWindow>.Context context, Iterable<Tuple2<String, Integer>> elements, Collector<String> out) throws Exception {
                        String key = tuple.getField(0);
                        String window = context.window().toString();
                        long currentWatermark = context.currentWatermark();
                        long currentProcessingTime = context.currentProcessingTime();
                        int i = 0;
                        StringBuilder buffer = new StringBuilder();
                        for (Tuple2<String, Integer> element : elements) {
                            buffer.append(element);
                            i++;
                        }
                        out.collect("key:" + key + " window:" + window + " currentWatermark:" + currentWatermark + " currentProcessingTime:" + currentProcessingTime + " wordNum:" + i + " words:" + buffer);
                    }
                });

        sum.print().setParallelism(1);

        // execute program
        env.execute("WindowWordCount");
    }
}
